package today.wxp.blog.spark.structed.streaming.processor

import java.util.concurrent.TimeUnit

import org.apache.spark.sql.{ForeachWriter, Row, SparkSession}
import org.apache.spark.sql.streaming.Trigger
import today.wxp.blog.spark.structed.streaming.util.PlayInfoDeserializerWrapper

/**
  * 插入指标计算。
  * <p/>
  * Author   : wangxp
  * <p/>
  * DateTime : 2017/11/23 11:20
  */
object PlayIndexCalculationProcessor
{
  def main(args: Array[String]): Unit =
  {
    val spark = SparkSession
      .builder()
      .appName("PlayIndexCalculationProcessor")
      .getOrCreate()

    //import spark.implicits._

    val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "124.193.205.83:9092")
      .option("subscribe", "test0")
      .option("group.id", "test-consumer-group")
      //.option("startingOffsets", "earliest")
      .load()

    spark.udf.register("deserialize", (topic: String, bytes: Array[Byte]) =>
      PlayInfoDeserializerWrapper.deSerializer.deserialize(topic, bytes)
    )

    df.selectExpr("""deserialize(topic, value) AS play_info""").createTempView("t_platy_info")

    val flatTypeInfoSQL =
      """SELECT uid, 1.0 * SUM(play_flag) / SUM(all_flag) AS calc_rate
        |FROM
        |	(
        |	SELECT uid, CASE play_type WHEN 'play' THEN 1 ELSE 0 END AS play_flag, 1 AS all_flag
        |	FROM
        |		(
        |		SELECT uid, play_item.json_value.type AS play_type
        |		FROM
        |			(
        |			SELECT CONCAT(play_info.cust_id, play_info.client_id, play_info.instance_id, play_info.packet.target_data.target_id) AS uid
        |			 	, play_info.packet.target_data.item_list AS play_items
        |			FROM t_platy_info
        |			) AS t_play_items LATERAL VIEW EXPLODE(play_items) v_play_items AS play_item
        |		)
        |		AS t_flat_play_info
        |	WHERE play_type IN ('play', 'beforePlay')
        |	) AS t_flag_result
        |GROUP BY uid""".stripMargin

    val result = spark.sql(flatTypeInfoSQL)
      .writeStream
      .outputMode("complete")
      .foreach(new ForeachWriter[Row]
      {
        override def process(value: Row): Unit = println(value)

        override def close(errorOrNull: Throwable): Unit =
        {}

        override def open(partitionId: Long, version: Long): Boolean = true
      })
      .format("console")
      .trigger(Trigger.ProcessingTime(5, TimeUnit.SECONDS))
      .start()

    result.awaitTermination()
  }
}
